package com.example.generator;

import com.alibaba.fastjson.JSON;
import com.example.model.CanalMessage;
import com.example.model.OGGMessage;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.*;
import java.util.concurrent.TimeUnit;

public class DataGenerator {
    private final KafkaProducer<String, String> producer;
    private final Random random = new Random();
    private final String[] operations = {"INSERT", "UPDATE", "DELETE"};
    private final String[] tables = {"users", "orders", "products"};

    public DataGenerator(Properties kafkaProps) {
        this.producer = new KafkaProducer<>(kafkaProps);
    }

    // 生成Canal格式数据
    public void generateCanalData(String topic, int count) {
        for (int i = 0; i < count; i++) {
            CanalMessage message = new CanalMessage();
            String table = tables[random.nextInt(tables.length)];
            
            message.setDatabase("test_db");
            message.setTable(table);
            message.setType(operations[random.nextInt(operations.length)]);
            message.setEs(System.currentTimeMillis());
            message.setTs(System.currentTimeMillis());
            
            // 生成数据
            List<Map<String, Object>> data = new ArrayList<>();
            Map<String, Object> row = new HashMap<>();
            row.put("id", random.nextInt(10000));
            row.put("name", "测试数据_" + UUID.randomUUID().toString().substring(0, 8));
            row.put("create_time", new Date().toString());
            data.add(row);
            
            message.setData(data);
            message.setPkNames(Collections.singletonList("id"));
            
            // 发送到Kafka
            String messageJson = JSON.toJSONString(message);
            producer.send(new ProducerRecord<>(topic, messageJson));
            
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    // 生成OGG格式数据
    public void generateOGGData(String topic, int count) {
        for (int i = 0; i < count; i++) {
            OGGMessage message = new OGGMessage();
            String table = tables[random.nextInt(tables.length)];
            
            message.setOp_type(operations[random.nextInt(operations.length)]);
            message.setOp_ts(new Date().toString());
            message.setCurrent_ts(String.valueOf(System.currentTimeMillis()));
            message.setPos("oracle.log.123456789");
            message.setTable(table);
            
            // 生成数据
            Map<String, Object> data = new HashMap<>();
            int id = random.nextInt(10000);
            data.put("id", id);
            data.put("name", "测试数据_" + UUID.randomUUID().toString().substring(0, 8));
            data.put("create_time", new Date().toString());
            
            message.setAfter(data);
            message.setPrimary_keys(Collections.singletonMap("id", id));
            
            // 发送到Kafka
            String messageJson = JSON.toJSONString(message);
            producer.send(new ProducerRecord<>(topic, messageJson));
            
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public void close() {
        producer.close();
    }
} 